package com.cmc6.websocket.service.impl;

import cn.hutool.core.collection.CollUtil;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.cmc6.common.exception.BaseBizCodeEnum;
import com.cmc6.common.model.dto.NotEmptyIdSet;
import com.cmc6.common.model.entity.BaseEntityThree;
import com.cmc6.common.model.entity.BaseEntityTwo;
import com.cmc6.common.model.enums.RequestCategoryEnum;
import com.cmc6.common.util.KafkaUtil;
import com.cmc6.common.util.MyJwtUtil;
import com.cmc6.websocket.config.WebSocketApplicationRunnerConfig;
import com.cmc6.websocket.mapper.SocketMapper;
import com.cmc6.websocket.model.entity.SocketDO;
import com.cmc6.websocket.service.SocketService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.Date;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

@Service
@Slf4j
public class SocketServiceImpl extends ServiceImpl<SocketMapper, SocketDO> implements SocketService {

    /**
     * {@link com.cmc6.websocket.service.WebSocketHelperService#reg}
     * 新增用户在线信息
     */
    @Override
    public void online(SocketDO entity) {

        // 注意：这个需要在这里进行设置
        entity.setServer(WebSocketApplicationRunnerConfig.socketSufKey);

        save(entity); // 保存到数据库

    }

    /**
     * 移除用户在线信息：通过 socketId，并且进行 socket通知
     */
    @Override
    public void offlineAndNoticeBySocketIdSetAndUserId(Set<Long> socketIdSet, Long userId,
        RequestCategoryEnum requestCategoryEnum) {

        offlineBySocketIdSet(socketIdSet); // 先操作数据库

        // 移除 redis中的 jwtHash
        MyJwtUtil.removeJwtHashByRequestCategoryOrJwtHash(userId, requestCategoryEnum, null, null);

        // 并且给 消息中间件推送，进行下线操作
        KafkaUtil.forcedOffline(socketIdSet, null);

    }

    /**
     * 移除用户在线信息：通过 socketId
     */
    @Override
    public void offlineBySocketIdSet(Set<Long> socketIdSet) {
        if (CollUtil.isEmpty(socketIdSet)) {
            return;
        }
        lambdaUpdate().in(SocketDO::getId, socketIdSet).eq(BaseEntityThree::getEnableFlag, true)
            .set(BaseEntityThree::getEnableFlag, false).set(BaseEntityTwo::getUpdateTime, new Date())
            .update(); // enableFlag：连接中/断开连接
    }

    /**
     * 移除用户在线信息：通过 userIdSet
     */
    @Override
    public void offlineByUserIdSet(Set<Long> userIdSet) {
        if (CollUtil.isEmpty(userIdSet)) {
            return;
        }
        lambdaUpdate().in(SocketDO::getUserId, userIdSet).eq(BaseEntityThree::getEnableFlag, true)
            .set(BaseEntityThree::getEnableFlag, false).set(BaseEntityTwo::getUpdateTime, new Date())
            .update(); // enableFlag：连接中/断开连接
    }

    /**
     * 移除用户在线信息：通过 userIdSet
     */
    @Override
    public void offlineAndNoticeByUserIdSet(Set<Long> userIdSet) {

        offlineByUserIdSet(userIdSet);

        // 移除 redis中的 jwtHash
        for (Long item : userIdSet) {
            MyJwtUtil.removeJwtHashByRequestCategoryOrJwtHash(item, null, null, null);
        }

        // 并且给 消息中间件推送，进行下线操作
        KafkaUtil.forcedOffline(null, userIdSet);
    }

    /**
     * 启动项目 或者springboot销毁时：断开当前 netty-socket服务器所有连接
     */
    @Override
    public void offlineAllForCurr() {
        log.info("下线数据库的 socket连接，条件：server = {}", WebSocketApplicationRunnerConfig.socketSufKey);
        lambdaUpdate().eq(SocketDO::getServer, WebSocketApplicationRunnerConfig.socketSufKey)
            .eq(BaseEntityThree::getEnableFlag, true).set(BaseEntityThree::getEnableFlag, false).update();
    }

    /**
     * 强退，通过 idSet
     */
    @Override
    @Transactional
    public String offlineByIdSet(NotEmptyIdSet notEmptyIdSet) {

        offlineBySocketIdSet(notEmptyIdSet.getIdSet());

        List<SocketDO> socketDbList = lambdaQuery().in(BaseEntityTwo::getId, notEmptyIdSet.getIdSet())
            .select(SocketDO::getJwtHash, SocketDO::getUserId).groupBy(SocketDO::getJwtHash).list();
        if (socketDbList.size() != 0) {
            for (SocketDO item : socketDbList) {
                // 清理：jwtUser set里面的 jwtHash，以及 jwtHash
                MyJwtUtil.removeJwtHashByRequestCategoryOrJwtHash(item.getUserId(), null, item.getJwtHash(), null);
            }
        }

        // 并且给 消息中间件推送，进行下线操作
        KafkaUtil.loginExpired(notEmptyIdSet.getIdSet());

        return BaseBizCodeEnum.API_RESULT_OK.getMsg();
    }

    /**
     * 全部强退
     */
    @Override
    @Transactional
    public String offlineAll() {

        MyJwtUtil.removeAllJwtHash();

        List<SocketDO> socketDbList =
            lambdaQuery().eq(BaseEntityThree::getEnableFlag, true).select(BaseEntityTwo::getId).list();
        Set<Long> socketIdSet = socketDbList.stream().map(BaseEntityTwo::getId).collect(Collectors.toSet());

        if (socketIdSet.size() == 0) {
            return BaseBizCodeEnum.API_RESULT_OK.getMsg();
        }

        lambdaUpdate().eq(BaseEntityThree::getEnableFlag, true).set(BaseEntityThree::getEnableFlag, false)
            .set(BaseEntityTwo::getUpdateTime, new Date()).update(); // enableFlag：连接中/断开连接

        // 并且给 消息中间件推送，进行下线操作
        KafkaUtil.loginExpired(socketIdSet);

        return BaseBizCodeEnum.API_RESULT_OK.getMsg();
    }

    /**
     * 在注销用户的时候，通过 userIdSet，下线用户
     */
    @Override
    public void offlineByUserIdSetForDeleteUser(Set<Long> userIdSet) {

        offlineByUserIdSet(userIdSet);

        // 并且给 消息中间件推送，进行下线操作
        KafkaUtil.delAccount(userIdSet);

    }
}




